package com.zte.claa.inficombo.csa.app.service.resrpt.impl;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.model.dev.DevTypeModel;
import com.zte.claa.inficombo.csa.app.model.dev.ResIndexDataModel;
import com.zte.claa.inficombo.csa.app.model.dev.ResourceShadow;
import com.zte.claa.inficombo.csa.app.model.redis.DevMeteData;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.resrpt.ResIndexDataRptService;
import com.zte.claa.inficombo.csa.app.service.rptparse.ParserContainer;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;

@Service
public class ResIndexDataRptServiceImpl implements ResIndexDataRptService, InitializingBean, DisposableBean {
	// 日志记录
	private final Logger logger = LoggerFactory.getLogger(this.getClass());		
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private ParserContainer parseContainer;
	
	/**
	 * 持久化线程最大数目.
	 */
	@Value("${csa.res.index.data.threadnum}")
	private int threadNum = 5;
	
	/**
	 * 线程池队列最大长度.
	 */
	@Value("${csa.res.index.data.queuesize}")
	private int maxqueuesize = 1000;
	
	// 是否过滤重复数据.
	@Value("${csa.res.index.data.filtersamedata}")
	private boolean filtersamedata = false;
	
	// 重复数据过滤检查间隔（单位：秒）
	@Value("${csa.res.index.data.samedatacheckperiod}")
	private int samedatacheckperiod = 300;
	
	// 数据持久化和发布处理线程池...
	private ExecutorService execSvr;
	
	@Override
	public void afterPropertiesSet() throws Exception {
		this.execSvr = new ThreadPoolExecutor(threadNum, threadNum+3, 60, TimeUnit.MILLISECONDS, 
				new LinkedBlockingQueue<Runnable>(maxqueuesize));	
	}

	@Override
	public void destroy() throws Exception {
		if(this.execSvr != null){
			this.execSvr.shutdown();
			this.execSvr = null;
		}
	}

	@Override
	public boolean submitResIndexData(DevMeteData meteData) {
		try{
			this.execSvr.execute(new Runnable(){
				@Override
				public void run() {
					processResIndexDataRpt(meteData);
				}			
			});
		}catch(Exception e){
			logger.error("submitResIndexData failed: "+meteData, e);
			return false;
		}		
		return true;
	}

	/**
	 * Res的指标数据处理:
	 * 注：meteData中的deveui是必须输入的.
	 * 
	 * 1）查询缓存，补充projecteid、devtype等类型；
	 * 2）持久化保存；
	 * 3）数据发布；
	 * 
	 * @param meteData
	 */
	private void processResIndexDataRpt(DevMeteData meteData) {
		if(meteData == null || meteData.getDeveui() == null || meteData.getDeveui().equals("")){
			logger.warn("DevEui is null!");
			return;
		}
		long bt = System.currentTimeMillis();
		String devEui = meteData.getDeveui();
		// 本服务仅针对逻辑资源，因此这里只查询逻辑资源... 
		ResourceShadow res = this.cacheManager.getRes(devEui);
		if(res == null){
			logger.warn("Res instance NOT exist and data is discarded: {}, data={}", devEui, meteData);
			return;
		}
		DevTypeModel dtm = this.cacheManager.getDevTypeModel(res.getRestype());
		if(dtm == null || dtm.getBase() == null){
			logger.error("[*!!*]Can not get DevTypeModel for devtype="+res.getRestype());
			return;
		}
		// 检查是否要进行相同数据过滤.
		if(this.filtersamedata){
			if(this.isSameDataRpt(meteData)){
				return;
			}
		}	
		try {
			meteData.setDevname(res.getResname());
			meteData.setDevtype(res.getRestype());
			meteData.setProjectid(res.getProjectid());
			meteData.setPort("0");
			if (meteData.getRpttime() == null) {
				meteData.setRpttime(new Timestamp(System.currentTimeMillis()));
			}
			// 根据模型过滤指标字段，避免多报的数据影响入库.
			Map<String, Object> orginDataMap = meteData.getMdata();
			List<ResIndexDataModel> dmList = dtm.getResIndexDMList();
			if(dmList != null && orginDataMap != null){
				Map<String, Object> newDataMap = new HashMap<String, Object>();
				for(ResIndexDataModel dm : dmList){
					String fieldName = dm.getFieldname();
					Object fieldVal = orginDataMap.get(fieldName);
					if(fieldVal != null){
						newDataMap.put(fieldName, fieldVal);
					}					
				}
				meteData.setMdata(newDataMap);
			}			
			// 持久化数据.
			this.parseContainer.getPersisService().saveDevMeteData(meteData, dtm.getBase().getTbname());
			// 构造广播数据，进行redis发布...
			String pubMsg = JsonUtils.toJson(meteData);
			this.parseContainer.getUpdataPublisher().publishMessage(pubMsg);
			long et = System.currentTimeMillis();
			logger.info("[*^_^*]Publish res index metedata message ok: usetime(ms)={}, msg={}", (et-bt), pubMsg);
			// 如果需要过滤重复上报的话，则更新缓存.
			if(this.filtersamedata){
				// 更新缓存.
				this.cacheManager.putResNewIndexData(meteData);
			}	
		}catch(Exception e){
			logger.error("save red index data failed: "+meteData, e);
		}		
	}
	
	/**
	 * 检查本次上报的指标是否为相同数据上报（无变化）：
	 * 1）metedata各指标对应相同；
	 * 2）最近上报时间在指定时长范围内；
	 * 
	 * @param newData
	 * @return
	 */
	private boolean isSameDataRpt(DevMeteData newData) {
		try {
			DevMeteData oldData = this.cacheManager.getResNewIndexData(newData.getDeveui());
			if (oldData == null) {
				return false;
			}
			// 超过了检查时长范围，按新数据上报处理.
			if ((System.currentTimeMillis() - oldData.getRpttime().getTime()) >= (this.samedatacheckperiod * 1000L)) {
				return false;
			}
			Map<String, Object> oldMap = oldData.getMdata();
			Map<String, Object> newMap = newData.getMdata();
			if (oldMap == null || newMap == null) {
				return false;
			}
			Iterator<String> it = newMap.keySet().iterator();
			while (it.hasNext()) {
				String fk = it.next();
				Object newObj = newMap.get(fk);
				Object oldObj = oldMap.get(fk);
				if (newObj == null || oldObj == null) {
					return false;
				}
				if (!newObj.equals(oldObj)) {
					return false;
				}
			}
			return true;
		}catch(Exception e){
			logger.error("isSameDataRpt failed: deveui={}, err={}", newData.getDeveui(), e.getMessage());
		}
		return false;
	}
	
}
